package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
    TODO 获取指定分区的最大值--mapPartitionsWithIndext转换算子
     说明：
      1、分布式中无法保证第一个执行的分区就是第一个分区
      2、分布式本身就是没有顺序的
    因此：
      1、Spark 提供了一个算子，可以获取分区数据的同时获取分区编号:分区编号从0开始
      2、mapPartitionsWithIndex(第一个参数为分区编号，第二各位参数为分区的数据)

*/
object Spark_RDD_Pper_Transform_mapPartitionsWithIndex {
  def main(args: Array[String]): Unit = {


    //TODO 1、获得连接
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑
    // 获取指定分区的最大值--分布式中无法保证第一个执行的分区就是第一个分区，
    // 所以下面要使用mapPartitionsWithIndext

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    val value: RDD[Int] = rdd.mapPartitionsWithIndex(
      //3.1 这里参数列表第一个是分区的编号，第二个参数是数据本身
      (indext, iter) => {
        //3.2 我们要指定处理分区的数，所以要判断第一个参数，这里的分区是从0开始的
        if (indext == 1) {
          //3.3 这里要返回一个迭代器
          List(iter.max).iterator
        } else {
          //3.4 如果不是就返回一个空的
          Nil.iterator
        }
      }
    )
    //TODO 4、打印遍历
    value.collect().foreach(println)
    //TODO 2、关闭连接
    sc.stop()
  }
}